#!/usr/bin/env php
<?php
//$SOAR_DIR=getenv("SOAR_DIR");
//if(empty($SOAR_DIR)) $SOAR_DIR="/opt/soar";
$SOAR_DIR="../../../../..";
include "$SOAR_DIR/common/php/soar/soar.inc";

exit(main($argv));

function main($argv)
{/*{{{*/
    $daemon=false;
    if(isset($argv[1]))
    {
        if($argv[1] == "-d" || $argv[1] == "--daemon")
           $daemon=true; 
    }
    if(load_cnfg() == false)
        return 1;
    if(start($daemon) == false)
        return 1;
    return 0;
}/*}}}*/

function load_cnfg()
{/*{{{*/
    $cnfg_file=dirname(__FILE__)."/cnfg.inc";
    include $cnfg_file;

    if(empty(Cnfg::$workers))
    {
        printf("%s: Cnfg::workers not set\n", __FUNCTION__);
        return false;
    }
    if(Cnfg::$workers < 1 || Cnfg::$workers > 32)
        Cnfg::$workers=4;

    if(empty(Cnfg::$access['server_port']) ||
            Cnfg::$access['server_port'] < 1024)
    {
        printf("%s: Cnfg::access.server_port invalid\n", __FUNCTION__);
        return false;
    }
    if(empty(Cnfg::$access['monitor_port']) ||
            Cnfg::$access['monitor_port'] < 1024)
    {
        printf("%s: Cnfg::access.monitor_port invalid\n", __FUNCTION__);
        return false;
    }

    return true;
}/*}}}*/

class Share
{/*{{{*/
    private static $inst=null;

    private function __construct()
    {/*{{{*/
    }/*}}}*/

    public static function init()
    {/*{{{*/
        if(self::$inst == null)
        {
            self::$inst=new Share();
        }
    }/*}}}*/

    public static function getInstance()
    {/*{{{*/
        return self::$inst;
    }/*}}}*/

    public $stat_lock;
    public $stat;
    //array(
    //        'worker'=>array(
    //            array(
    //                'client_conn_num'=>,
    //                'client_call_num'=>,
    //                ),
    //            ),
    //        'svc'=>array(
    //            $name=>,
    //            ),
    //        );

    public $operate_lock;
    public $operate;
    //array(
    //        'worker'=>array(
    //            $status
    //            ),
    //        'quit'=>,
    //     );
}/*}}}*/

function start($daemon)
{/*{{{*/
    pcntl_signal(SIGPIPE, SIG_IGN);

    if($daemon)
    {/*{{{*/
        $pid=pcntl_fork();
        if($pid == -1)
        {
            printf("%s: daemonize fail", __METHOD__);
            return false;
        }
        if($pid > 0)
        {
            return true;
        }
        // Errlog
        if(Errlog::init(basename(__FILE__), Cnfg::$errlog['location'])
                == false)
        {
            printf("%s: init errlog fail", __METHOD__);
            return false;
        }
    }/*}}}*/

    // setuid
    if(posix_getuid() == 0)
    {/*{{{*/
        if(!empty(Cnfg::$user))
        {
            $user=posix_getpwnam(Cnfg::$user);
            if($user)
            {
                if(posix_setuid($user['uid']) == false)
                {
                    printf("Change to user %s fail\n", Cnfg::$user);
                    return false;
                }
                printf("Change to user %s\n", Cnfg::$user);
            }
            else
            {
                printf("User %s not found, keep root\n", Cnfg::$user);
            }
        }
        else
        {
            printf("User empty, keep root\n");
        }
    }/*}}}*/

    // 初始化共享信息
    Share::init();
    $share=Share::getInstance();

    $key=ftok(__FILE__, "s");
    if(($id=sem_get($key)) === false ||
            sem_remove($id) == false ||
            ($id=sem_get($key)) === false)
    {
        printf("%s: init sem(share) fail\n", __METHOD__);
        return false;
    }
    $share->stat_lock=$id;
    if(($id=shm_attach($key)) === false ||
            shm_remove($id) == false ||
            ($id=shm_attach($key)) === false)
    {
        printf("Error: init shm(share) fail\n");
        return false;
    }
    for($i=0; $i<Cnfg::$workers; ++$i)
        $stat['worker'][$i]=
            array('client_conn_num'=>0, 'client_call_num'=>0);
    $stat['svc']=array();
    shm_put_var($id, 0, $stat);
    $share->stat=$id;

    $key=ftok(__FILE__, "o");
    if(($id=sem_get($key)) === false ||
            sem_remove($id) == false ||
            ($id=sem_get($key)) === false)
    {
        printf("Error: init sem(operate) fail\n");
        return false;
    }
    $share->operate_lock=$id;
    if(($id=shm_attach($key)) === false ||
            shm_remove($id) == false ||
            ($id=shm_attach($key)) === false)
    {
        printf("Error: init shm(operate) fail\n");
        return false;
    }
    for($i=0; $i<Cnfg::$workers; ++$i)
        $operate['worker'][$i]='RUNNING';
    $operate['quit']=false;
    shm_put_var($id, 0, $operate);
    $share->operate=$id;

    if(($pids=worker::start()) === false)
        return false;

    Monitor::init();
    $monitor=Monitor::getInstance();
    return $monitor->start($pids);
}/*}}}*/

class Worker
{/*{{{*/
    public static function start()
    {/*{{{*/
        // 载入服务
        $svc_dir="./svc";
        $dh=opendir($svc_dir);
        while($inc=readdir($dh))
        {
            if(substr($inc, 0, 1) != "." &&
                    substr($inc, -4) == ".inc")
            {
                include_once "$svc_dir/$inc";
                printf("load $inc\n");
            }
        }

        $ep=new Ep();
        $sock=Sock::listen(Cnfg::$access['server_port']);
        if($sock === false)
        {
            printf("%s: open server_port(%d) fail",
                    __METHOD__, Cnfg::$access['server_port']);
            return false;
        }
        $server_slot_id=$ep->addListen($sock);
        if($server_slot_id === false)
        {
            Errlog::add("%s: Ep::addListen server fail", __METHOD__);
            return false;
        }
        printf("open server port:%d slot:%d\n",
                Cnfg::$access['server_port'], $server_slot_id);
        Errlog::add("%s: open server port %d",
                __METHOD__, Cnfg::$access['server_port']);

        $pids=array();
        for($i=0; $i<Cnfg::$workers; ++$i)
        {
            $pid=pcntl_fork();
            if($pid == -1)
            {
                printf("%s: fork workers fail\n", __FUNCTION__);
                return false;
            }
            if($pid == 0)
            {
                $worker=new Worker($i, $ep, $server_slot_id);
                printf("worker(%d) start\n", $i);
                $worker->loop();
                exit(0);
            }
            $pids[]=$pid;
        }

        return $pids;
    }/*}}}*/

    private $id;
    private $ep=null;
    private $listen_slots=array();    // 'server'=>

    private $client_conn_count=0;
    private $svc_call_count=array();

    public function __construct($id, $ep, $server_slot_id)
    {/*{{{*/
        $this->id=$id;
        $this->ep=$ep;
        $this->listen_slots['server']=$server_slot_id;
    }/*}}}*/

    private function loop()
    {/*{{{*/
        $UPDATE_STAT_INFO_INTERVAL=10;
        $CHECK_OPERATE_INFO_INTERVAL=2;

        $last_stat=0;
        $last_operate=0;
        $share=Share::getInstance();

        $quit=false;
        while(!$quit)
        {
            $active_list=$this->ep->poll();
            if($active_list === false)
            {
                Errlog::add("%s: ep::poll fail", __METHOD__);
            }
            else
            {
                foreach($active_list as $active)
                {
                    switch($active['type'])
                    {
                        case Ep::CONN:
                            self::conn($active);
                            break;
                        case Ep::RECV:
                            self::recv($active);
                            break;
                        case Ep::ABORT:
                        case Ep::TIMEOUT:
                            self::disconn($active);
                            break;
                    }
                }
            }

            $now=time();

            if($now-$last_stat > $UPDATE_STAT_INFO_INTERVAL)
            {/*{{{*/
                if($this->client_conn_count > 0)
                {
                    $client_call_count=0;

                    sem_acquire($share->stat_lock);
                    $stat=shm_get_var($share->stat, 0);

                    foreach($this->svc_call_count as $name=>$count)
                    {
                        if($count > 0)
                        {
                            if(array_key_exists($name, $stat['svc']))
                                $stat['svc'][$name]+=$count;
                            else
                                $stat['svc'][$name]=$count;
                            $client_call_count+=$count;
                        }
                    }
                    $stat['worker'][$this->id]
                        ['client_conn_num']+=$this->client_conn_count;
                    $stat['worker'][$this->id]
                        ['client_call_num']+=$client_call_count;

                    shm_put_var($share->stat, 0, $stat);
                    sem_release($share->stat_lock);

                    $this->client_conn_count=0;
                    $this->svc_call_count=array();
                }

                $last_stat=$now;
            }/*}}}*/

            if($now-$last_operate > $CHECK_OPERATE_INFO_INTERVAL)
            {/*{{{*/
                sem_acquire($share->operate_lock);
                $operate=shm_get_var($share->operate, 0);

                $quit=$operate['quit'];
                if($quit)
                {
                    $operate['worker'][$this->id]='STOP';
                    shm_put_var($share->operate, 0, $operate);
                    Errlog::add("%s: worker(%d) quit(%d)",
                            __METHOD__, $this->id, $quit);
                }

                sem_release($share->operate_lock);

                $last_operate=$now;
            }/*}}}*/
        }
    }/*}}}*/

    private function conn($active)
    {/*{{{*/
        $CONN_IDLE_TIMEOUT=120;

        $new_slot_id=$this->ep->add($active['new_sock']);
        if($new_slot_id === false)
        {
            Errlog::add("%s: worker(%d) Ep::add fail",
                    __METHOD__, $this->id);
            socket_close($active['new_sock']);
            return false;
        }
        $this->ep->setTimeout($new_slot_id, $CONN_IDLE_TIMEOUT);
        $this->ep->setWrap($new_slot_id);
        //$slot_info=$this->ep->getSlotInfo($new_slot_id);
        //$addr=$slot_info['peer_ip'].":".$slot_info['peer_port'];
        ++$this->client_conn_count;
        Errlog::add("%s: worker(%d) conn:%d",
                __METHOD__, $this->id, $new_slot_id);

        return true;
    }/*}}}*/

    private function recv($active)
    {/*{{{*/
        $response=self::callsvc($active['msg']);
        $res=$this->ep->send($active['slot_id'], $response);
        if($res == false)
        {
            Errlog::add("%s: worker(%d) send fail",
                    __METHOD__, $this->id); 
        }
    }/*}}}*/

    private function disconn($active)
    {/*{{{*/
        Errlog::add("%s: worker(%d) client %s %s, error:%d",
                __METHOD__, $this->id, $active['addr'],
                ($active['type']==Ep::ABORT)?"abort":"timeout",
                $active['error']);

        return true;
    }/*}}}*/

    private function callsvc($request_s)
    {/*{{{*/
        $request=json_decode($request_s, true);
        if($request == null)
        {
            Errlog::add("%s: worker(%d) request(%s) invalid",
                    __METHOD__, $this->id, substr($request_s, 0, 99)); 
            return json_encode(array('errcode'=>SOAR_ERR::MSG));
        }

        if(!isset($request['svc']) ||
                strlen($request['svc']) > Misc::NAME_SIZE)
        {
            Errlog::add("%s: worker(%d) svc(%s) invalid",
                    __METHOD__, $this->id,
                    isset($request['svc'])?$request['svc']:'null');
            return json_encode(array('errcode'=>SOAR_ERR::SVC_INVALID));
        }
        $svc=strtoupper($request['svc']);

        if(class_exists($svc))
        {
            $ins=new $svc();
            $response=$ins->call($request['paras']);
            if(array_key_exists($svc, $this->svc_call_count))
                ++$this->svc_call_count[$svc];
            else
                $this->svc_call_count[$svc]=1;
        }
        else
        {
            Errlog::add("%s: class(%s) not found",
                    __METHOD__, $svc);
            $response=array('errcode'=>SOAR_ERR::SVC_NOT_FOUND);
        }

        return json_encode($response);
    }/*}}}*/
}/*}}}*/

class Monitor
{/*{{{*/
    private static $inst=null;

    private function __construct()
    {/*{{{*/
    }/*}}}*/

    public static function init()
    {/*{{{*/
        if(self::$inst == null)
        {
            self::$inst=new Monitor();
        }
    }/*}}}*/

    public static function getInstance()
    {/*{{{*/
        return self::$inst;
    }/*}}}*/

    private $worker=array();

    private $quit=false;
    private $cmd_map=array(
            array(
                'name'=>"printworker", 'ab'=>"pwrk",
                'desc'=>"print client ip & port",
                'usage'=>"printclient(pwrk)"
                ),
            array(
                'name'=>"printservice", 'ab'=>"psvc",
                'desc'=>"print service status",
                'usage'=>"printservice(psvc)"
                ),
            array(
                'name'=>"shutdown", 'ab'=>"shutdown",
                'desc'=>"shutdown server",
                'usage'=>"shutdown"
                ),
            array(
                'name'=>"help", 'ab'=>"h",
                'desc'=>"",
                'usage'=>""
                )
                );

    public function start($worker_pids)
    {/*{{{*/
        foreach($worker_pids as $pid)
        {
            $this->worker[]=array('pid'=>$pid, 'alive'=>true);
        }

        $monitor_sock=Sock::listen(Cnfg::$access['monitor_port']);
        if($monitor_sock === false)
        {
            Errlog::add("%s: open monitor_port(%d) fail",
                    __METHOD__, Cnfg::$access['monitor_port']);
            return false;
        }

        $TIMEOUT=5;
        while(!$this->quit)
        {/*{{{*/
            self::scan_worker();

            $to=$TIMEOUT;
            $console_sock=Sock::accept($monitor_sock, $to);
            if($console_sock === false)
            {
                continue;
            }

            $console_quit=false;
            while(!$this->quit && !$console_quit)
            {
                self::scan_worker();

                $res=socket_recv($console_sock, $request, 500, 0);
                if($res === false || $res == 0)
                {
                    Errlog::add("%s: read console fail", __METHOD__);
                    $console_quit=true;
                    continue;
                }

                $request=trim($request);
                $pos=strpos($request, " ");
                if($pos === false)
                {
                    $cmd=$request;
                    $body="";
                }
                else
                {
                    $cmd=substr($request, 0, $pos);
                    $body=substr($request, $pos);
                    $body=ltrim($body);
                }
                $cmd=strtolower($cmd);
                //printf("cmd(%s)\n", $cmd);
                $cmdsize=strlen($cmd);
                if(0 < $cmdsize && $cmdsize <= Misc::NAME_SIZE) 
                {
                    $found=false;
                    foreach($this->cmd_map as $arr)
                    {
                        if($arr['name'] == $cmd || $arr['ab'] == $cmd)
                        {
                            $found=true;
                            $cmd=$arr['name'];
                            break;
                        }
                    }
                    if($found)
                    {
                        $func="CMD_FUNC_".$cmd;
                        $response=self::$func($body);
                    }
                    else
                    {
                        $response="command(".$cmd.") not found\n";
                    }
                }
                else
                {
                    $response="command invalid\n";
                }

                $res=socket_send($console_sock,
                        $response, strlen($response), 0);
                if($res === false || $res == 0)
                {
                    Errlog::add("%s: write console fail", __METHOD__);
                    $console_quit=true;
                    continue;
                }
            }
            socket_close($console_sock);
        }/*}}}*/

        Errlog::add("%s: shutdown", __METHOD__);
        return true;
    }/*}}}*/

    private function scan_worker()
    {/*{{{*/
        $share=Share::getInstance();

        for($i=0; $i<Cnfg::$workers; ++$i)
        {
            $worker=&$this->worker[$i];
            if($worker['alive'])
            {
                $pid=pcntl_waitpid($worker['pid'], $status, WNOHANG);
                //printf("$pid: $status\n");
                if($pid > 0)
                {
                    sem_acquire($share->operate_lock);
                    $operate=shm_get_var($share->operate, 0);
                    if($operate['worker'][$i] == 'RUNNING')
                    {
                        $operate['worker'][$i]='ABORT';
                        shm_put_var($share->operate, 0, $operate);
                        Errlog::add("%s: worker(%d) abort",
                                __METHOD__, $i);
                    }
                    sem_release($share->operate_lock);

                    $worker['alive']=false; 
                }
            }
        }
    }/*}}}*/

    private function cmd($request)
    {/*{{{*/
        $found=false;
        foreach(self::$cmd_map as $arr)
        {
            if($arr['name'] == $cmd || $arr['ab'] == $cmd)
            {
                $found=true;
                $cmd=$arr['name'];
                break;
            }
        }
        if($found)
        {
            $func="CMD_FUNC_".$cmd;
            $response=self::$func($body);
        }
        else
        {
            $response="command(".$cmd.") not found\n";
        }

        return $response;
    }/*}}}*/

    private function CMD_FUNC_PRINTWORKER($request)
    {/*{{{*/
        $buf=array();
        $share=Share::getInstance();
        $response=
            "Worker         Connected        Calls          Status\n".
            "------         ---------        -----          ------\n";

        sem_acquire($share->stat_lock);
        $stat=shm_get_var($share->stat, 0);
        for($i=0; $i<Cnfg::$workers; ++$i)
        {
            $worker=$stat['worker'][$i];
            $buf[$i]=sprintf("%6d         %9d    %9d",
                    $i,
                    $worker['client_conn_num'],
                    $worker['client_call_num']);
        }
        sem_release($share->stat_lock);

        sem_acquire($share->operate_lock);
        $operate=shm_get_var($share->operate, 0);
        for($i=0; $i<Cnfg::$workers; ++$i)
        {
            $buf[$i].=sprintf("          %s", $operate['worker'][$i]);
        }
        sem_release($share->operate_lock);

        for($i=0; $i<Cnfg::$workers; ++$i)
        {
            $response.=$buf[$i]."\n";
        }

        return $response;
    }/*}}}*/

    private function CMD_FUNC_PRINTSERVICE($request)
    {/*{{{*/
        $share=Share::getInstance();

        $response=
            "Service                         Calls\n".
            "-------                         ---------\n";
        sem_acquire($share->stat_lock);
        $stat=shm_get_var($share->stat, 0);
        foreach($stat['svc'] as $name=>$num)
        {
            $response.=sprintf("%-30s  %9d\n",
                    $name, $num);
        }
        sem_release($share->stat_lock);

        return $response;
    }/*}}}*/

    private function CMD_FUNC_SHUTDOWN($request)
    {/*{{{*/
        $share=Share::getInstance();

        sem_acquire($share->operate_lock);
        $operate=shm_get_var($share->operate, 0);
        $operate['quit']=true;
        shm_put_var($share->operate, 0, $operate);
        sem_release($share->operate_lock);

        sleep(5);

        sem_acquire($share->operate_lock);
        $operate=shm_get_var($share->operate, 0);
        $finish=true;
        for($i=0; $i<Cnfg::$workers; ++$i)
        {
            $worker=$operate['worker'][$i];
            if($worker == 'RUNNING')
            {
                $finish=false;
                break;
            }
        }
        sem_release($share->operate_lock);

        if($finish)
        {
            $response="Shutdown complete\nDisconnect\n";
            $this->quit=true;
        }
        else
        {
            $response="Shutdown not complete, try again\n";
        }

        return $response;
    }/*}}}*/

    private function CMD_FUNC_HELP($request)
    {/*{{{*/
        if(empty($request))
        {
            $response="help [command]\ncommands:\n";
            foreach($this->cmd_map as $arr)
            {
                $response.="    ".$arr['name']."\n";
            }
            return $response;
        }

        $request=strtolower($request);
        foreach($this->cmd_map as $arr)
        {
            if($arr['name'] == $request || $arr['ab'] == $request)
            {
                $response="Description: ".$arr['desc']."\n".
                    "Usage: ".$arr['usage']."\n";
                return $response;
            }
        }
        $response="Command not found, type help\n";
        return $response;
    }/*}}}*/
}/*}}}*/
?>
